package com.myorg.util;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkHelper {
	private static final Logger logger = LoggerFactory.getLogger(ZkHelper.class);
	/**
	 * @return
	 * @throws UnknownHostException
	 */
	public static String createZnode() throws UnknownHostException {
		// connect to Zookeeper and create a path so that search slave can synchronize
		InetAddress ip = InetAddress.getLocalHost();
		final String uniquePath = Utility.getUniqueStrAcrossMultiJVM(ip.getHostAddress());
		try {
			ZkConnector zkc = new ZkConnector();
			zkc.connect(ConstantsToBeExternalized.ZOOKEEPER_HOST);
			ZooKeeper zk = zkc.getZooKeeper();
			logger.info("syncPath to be created is /" + uniquePath);
			zk.exists("/" + uniquePath, new Watcher() { // Anonymous Watcher
						@Override
						public void process(WatchedEvent event) {
							// check for event type NodeCreated
							boolean isNodeCreated = event.getType().equals(EventType.NodeCreated);
							// verify if this is the defined znode
							boolean isMyPath = event.getPath().equals(uniquePath);
							if (isNodeCreated && isMyPath) {
								logger.info("syncPath is created:" + uniquePath);
							} else {
								logger.error("syncPath is NOT created:" + uniquePath);
							}
						}
					});
			zk.create("/" + uniquePath, null, // TODO, data to be attached.
					Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			zk.close();
		} catch (IOException e1) {
			e1.printStackTrace();
			throw new RuntimeException(e1);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
			throw new RuntimeException(e1);
		} catch (KeeperException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
		return uniquePath;
	}

	/**
	 * @param syncPath
	 * @throws UnknownHostException
	 */
	public static void deleteZnode(String syncPath) throws UnknownHostException {
		String path = "/" + syncPath;
		try {
			ZkConnector zkc = new ZkConnector();
			zkc.connect(ConstantsToBeExternalized.ZOOKEEPER_HOST);
			ZooKeeper zk = zkc.getZooKeeper();

			zk.delete(path, -1);
			zk.close();
		} catch (KeeperException.NoNodeException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (KeeperException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
	}
}
